package com.tuozixuan.api.impl;

import com.tuozixuan.api.WebFluxDemoApi;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.MalformedURLException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.time.LocalTime;

@RestController
public class WebFluxDemoApiImpl implements WebFluxDemoApi {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());


    public Flux<Integer> getIntegerFlux() {
        return Flux.range(1, 10)
                .delayElements(Duration.ofMillis(500));
    }

    @Override
    public Mono<String> getString() {
        return Mono.just("Hello, WebFlux!");
    }

    @Override
    public Flux<String> getStringFlux() {
        return Flux.just("java", "python", "golang");
    }

    @Override
    public Flux<String> getTextEventStream() {
        return Flux.interval(Duration.ofSeconds(1)).map(sequence -> "SSE Event #" + sequence + " - " + LocalTime.now());
    }

    @Override
    public Flux<byte[]> getByteArrayStream() {
        logger.info("getByteArrayStream start========================");
        return Flux.interval(Duration.ofMillis(200))
                .map(sequence -> {
                    String data = "Data chunk #" + sequence + "\n";
                    logger.info("data:{}", data);
                    return data.getBytes(StandardCharsets.UTF_8);
                })
                .take(10);
    }

    @Override
    public Flux<ByteBuffer> getByteBufferStream() {
        return Flux.interval(Duration.ofMillis(200))
                .map(sequence -> {
                    String data = "Buffer chunk #" + sequence + "\n";
                    logger.info("data:{}", data);
                    return ByteBuffer.wrap(data.getBytes(StandardCharsets.UTF_8));
                })
                .take(5);
    }

    @Override
    public Flux<byte[]> getAudioStream() throws MalformedURLException {

        logger.info("getAudioStream Start");
        Path audioDir = Paths.get("D:\\window\\Downloads");
        Path filePath = audioDir.resolve("c467e685ed2b63f2.mp3").normalize();
        Resource resource = new UrlResource(filePath.toUri());

        return DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 4096)
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer); // 释放缓冲区
                    return bytes;
                });
    }

    @Override
    public Flux<byte[]> getRequestAudioStream() throws MalformedURLException {

        logger.info("getRequestAudioStream Start");
        return WebClient.create("http://localhost:8085")
                .get()
                .uri("/webFluxDemo/getAudioStream")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(byte[].class)
                .doOnNext(data -> {
                    System.out.println("Received: " + new String(data, StandardCharsets.UTF_8));
                })
                .doOnComplete(() -> System.out.println("Completed"))
                .doOnError(System.out::println);
    }

    @Override
    public Flux<byte[]> getByteArrayOctetStream() {
        logger.info("getByteArrayOctetStream start========================");
        return Flux.interval(Duration.ofMillis(200))
                .map(sequence -> {
                    String data = "Data chunk #" + sequence + "\n";
                    logger.info("data:{}", data);
                    return data.getBytes(StandardCharsets.UTF_8);
                })
                .take(10);
    }

    @Override
    public Flux<byte[]> getAudioOctetStream() throws MalformedURLException {

        logger.info("getAudioOctetStream Start");
        Path audioDir = Paths.get("D:\\window\\Downloads");
        Path filePath = audioDir.resolve("c467e685ed2b63f2.mp3").normalize();
        Resource resource = new UrlResource(filePath.toUri());

        return DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 4096)
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer); // 释放缓冲区
                    return bytes;
                });
    }
}
